package com.talent.service.iot.mqtt.service;

import com.talent.service.iot.mqtt.model.TlMqttSession;
import com.talent.service.iot.mqtt.model.entity.TlSubClient;
import com.talent.service.iot.mqtt.model.entity.TlTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @ClassName：DefaultSubscriptionServiceImpl
 * @Author: hszhou
 * @Date: 2025/1/8 9:24
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@Service
@ConditionalOnProperty(prefix = "mqtt", name = "store-strategy", havingValue = StoreStrategy.LOCAL)
public class LocalMqttSubscriptionServiceImpl implements IMqttSubscriptionService,ITopicService{

    private static final Logger logger = LoggerFactory.getLogger(LocalMqttSubscriptionServiceImpl.class);

    public static final ConcurrentHashMap<String, Set<TlSubClient>> TOPIC_TO_CLIENT = new ConcurrentHashMap<>();


    @Autowired
    private IMqttSessionService sessionService;

    @Override
    public Mono<TlSubClient> subscribe(TlSubClient clientSub) {
        String topic = clientSub.getTopic();
        TOPIC_TO_CLIENT.compute(topic,(k,v)->{
            if(v==null){
              v=new HashSet<>();
            }
            v.add(clientSub);
            return v;
        });

        return Mono.justOrEmpty(clientSub);
    }

    @Override
    public Mono<TlSubClient> unsubscribe(String clientId, String topic) {
        AtomicReference<TlSubClient> subClient= new AtomicReference<>();
        TOPIC_TO_CLIENT.computeIfPresent(topic, (k, v) -> {
            Iterator<TlSubClient> iterator = v.iterator();
            while (iterator.hasNext()) {
                subClient.set(iterator.next());
                if (subClient.get().getClientId().equals(clientId)) {
                    iterator.remove();
                }
            }
            return v;
        });
        return Mono.justOrEmpty(subClient.get());
    }

    @Override
    public Flux<TlSubClient> find(String topic) {
        return Flux.push(fluxSink-> TOPIC_TO_CLIENT.forEach((k, v)->{
            if (match(topic,k)) {
              v.forEach(fluxSink::next);
            }
        }));
//        Set<TlSubClient> rs = new HashSet<>();
//        TOPIC_TO_CLIENT.forEach((k,v)->{
//            if (match(topic,k)) {
//                rs.addAll(v);
//            }
//        });
//        return Flux.fromIterable(rs);


    }

    @Override
    public Mono<Void> clear(String clientId) {
        //获取这个客户端所有订阅的主题
        Mono<TlMqttSession> tlMqttSession = sessionService.find(clientId);

        tlMqttSession.subscribe(session -> {
            Set<TlTopic> topics = session.getTopics();
            for (TlTopic topic : topics) {
                TOPIC_TO_CLIENT.computeIfPresent(topic.getName(),(k,v)->{
                    Iterator<TlSubClient> iterator = v.iterator();
                    while (iterator.hasNext()){
                        TlSubClient next = iterator.next();
                        if(next.getClientId().equals(clientId)){
                            iterator.remove();
                        }
                    }
                    return v;
                });
            }
        });

        return Mono.empty();


    }
}
